package com.lengxf.flink.sink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;

public class SinkKafka {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        DataGeneratorSource<String> source = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) aLong -> "Number" + aLong,
                100,
                RateLimiterStrategy.perSecond(1),
                Types.STRING
        );
        //数据源消息
        DataStreamSource<String> stringDataStreamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "data-generator");
        //相对于kafka flink为生产者
        //构建kafka 的发送
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                //指定kafka服务
                .setBootstrapServers("localhost:7092,localhost:8092,localhost:9092")
                //指定序列化器
                .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                        .setTopic("foo3")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                //一致性级别
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                //精准一次的事务前缀
                .setTransactionalIdPrefix("trx-")
                //事务超时时间
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
                .build();
        stringDataStreamSource.sinkTo(kafkaSink);
        env.execute();
    }

}
